package megrez.keycloak.providers.mqtt;

import java.nio.charset.StandardCharsets;

import org.apache.commons.lang.StringUtils;
import org.eclipse.paho.client.mqttv3.IMqttClient;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import lombok.extern.slf4j.Slf4j;
import megrez.keycloak.common.KeycloakEmbeddedException;

/**
 * mqtt client实现
 * 
 * @author Lucky Yang
 * @since 2.6.9
 */
@Slf4j
public class ClientMqttService implements MqttService {
    private IMqttClient client;

    public void initClient(MqttConfigProperties properties) {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setAutomaticReconnect(properties.isAutomaticReconnect());
        options.setCleanSession(properties.isCleanSession());
        options.setConnectionTimeout(properties.getConnectionTimeout());
        options.setKeepAliveInterval(properties.getKeepAliveInterval());
        String username = properties.getUsername();
        if (StringUtils.isNotBlank(username)) {
            options.setUserName(username);
        }
        String password = properties.getPassword();
        if (StringUtils.isNotBlank(password)) {
            options.setPassword(password.toCharArray());
        }

        try {
            this.client = new MqttClient(properties.getBroker(), properties.getClientId(), new MemoryPersistence());
            this.client.connect(options);
        } catch (MqttException e) {
            throw new KeycloakEmbeddedException(e);
        }
    }

    @Override
    public void close() {
        // 不需要实现
    }

    private IMqttClient getMqttClient() {
        if (this.client == null) {
            throw new KeycloakEmbeddedException("MQTT Client is null, may be due to previous creation failure");
        }

        return this.client;
    }

    @Override
    public void connect() throws MqttException {
        this.getMqttClient().connect();
    }

    @Override
    public void connect(MqttConnectOptions options) throws MqttException {
        this.getMqttClient().connect(options);
    }

    @Override
    public IMqttToken connectWithResult(MqttConnectOptions options) throws MqttException {
        return this.getMqttClient().connectWithResult(options);
    }

    @Override
    public void disconnect() throws MqttException {
        this.getMqttClient().disconnect();
    }

    @Override
    public void disconnect(long quiesceTimeout) throws MqttException {
        this.getMqttClient().disconnect(quiesceTimeout);
    }

    @Override
    public void disconnectForcibly() throws MqttException {
        this.getMqttClient().disconnectForcibly();
    }

    @Override
    public void disconnectForcibly(long disconnectTimeout) throws MqttException {
        this.getMqttClient().disconnectForcibly(disconnectTimeout);
    }

    @Override
    public void disconnectForcibly(long quiesceTimeout, long disconnectTimeout) throws MqttException {
        this.getMqttClient().disconnectForcibly(quiesceTimeout, disconnectTimeout);
    }

    @Override
    public void subscribe(String topicFilter) throws MqttException {
        this.getMqttClient().subscribe(topicFilter);
    }

    @Override
    public void subscribe(String[] topicFilters) throws MqttException {
        this.getMqttClient().subscribe(topicFilters);
    }

    @Override
    public void subscribe(String topicFilter, int qos) throws MqttException {
        this.getMqttClient().subscribe(topicFilter, qos);
    }

    @Override
    public void subscribe(String[] topicFilters, int[] qos) throws MqttException {
        this.getMqttClient().subscribe(topicFilters, qos);
    }

    @Override
    public void subscribe(String topicFilter, IMqttMessageListener messageListener)
            throws MqttException {
        this.getMqttClient().subscribe(topicFilter, messageListener);
    }

    @Override
    public void subscribe(String[] topicFilters, IMqttMessageListener[] messageListeners) throws MqttException {
        this.getMqttClient().subscribe(topicFilters, messageListeners);
    }

    @Override
    public void subscribe(String topicFilter, int qos, IMqttMessageListener messageListener) throws MqttException {
        this.getMqttClient().subscribe(topicFilter, qos, messageListener);
    }

    @Override
    public void subscribe(String[] topicFilters, int[] qos, IMqttMessageListener[] messageListeners)
            throws MqttException {
        this.getMqttClient().subscribe(topicFilters, qos, messageListeners);
    }

    @Override
    public IMqttToken subscribeWithResponse(String topicFilter) throws MqttException {
        return this.getMqttClient().subscribeWithResponse(topicFilter);
    }

    @Override
    public IMqttToken subscribeWithResponse(String topicFilter, IMqttMessageListener messageListener)
            throws MqttException {
        return this.getMqttClient().subscribeWithResponse(topicFilter, messageListener);
    }

    @Override
    public IMqttToken subscribeWithResponse(String topicFilter, int qos) throws MqttException {
        return this.getMqttClient().subscribeWithResponse(topicFilter, qos);
    }

    @Override
    public IMqttToken subscribeWithResponse(String topicFilter, int qos, IMqttMessageListener messageListener)
            throws MqttException {
        return this.getMqttClient().subscribeWithResponse(topicFilter, qos, messageListener);
    }

    @Override
    public IMqttToken subscribeWithResponse(String[] topicFilters) throws MqttException {
        return this.getMqttClient().subscribeWithResponse(topicFilters);
    }

    @Override
    public IMqttToken subscribeWithResponse(String[] topicFilters, IMqttMessageListener[] messageListeners)
            throws MqttException {
        return this.getMqttClient().subscribeWithResponse(topicFilters, messageListeners);
    }

    @Override
    public IMqttToken subscribeWithResponse(String[] topicFilters, int[] qos) throws MqttException {
        return this.getMqttClient().subscribeWithResponse(topicFilters, qos);
    }

    @Override
    public IMqttToken subscribeWithResponse(String[] topicFilters, int[] qos, IMqttMessageListener[] messageListeners)
            throws MqttException {
        return this.getMqttClient().subscribeWithResponse(topicFilters, qos, messageListeners);
    }

    @Override
    public void unsubscribe(String topicFilter) throws MqttException {
        this.getMqttClient().unsubscribe(topicFilter);
    }

    @Override
    public void unsubscribe(String[] topicFilters) throws MqttException {
        this.getMqttClient().unsubscribe(topicFilters);
    }

    @Override
    public void publish(String topic, byte[] payload, int qos, boolean retained)
            throws MqttException {
        if (log.isDebugEnabled()) {
            log.info("Pulish message to Mqtt broker: topic={}, qos={}, retained={}, payload={}", topic,
                    qos,
                    retained, new String(payload, StandardCharsets.UTF_8));
        }
        this.getMqttClient().publish(topic, payload, qos, retained);
    }

    @Override
    public void publish(String topic, MqttMessage message) throws MqttException {
        if (log.isDebugEnabled()) {
            log.info("Pulish message to Mqtt broker: topic={}, qos={}, retained={}, payload={}", topic,
                    message.getQos(),
                    message.isRetained(), message.toString());
        }
        this.getMqttClient().publish(topic, message);

    }

    @Override
    public void setCallback(MqttCallback callback) {
        this.getMqttClient().setCallback(callback);

    }

    @Override
    public MqttTopic getTopic(String topic) {
        return this.getMqttClient().getTopic(topic);
    }

    @Override
    public boolean isConnected() {
        return this.getMqttClient().isConnected();
    }

    @Override
    public String getClientId() {
        return this.getMqttClient().getClientId();
    }

    @Override
    public String getServerURI() {
        return this.getMqttClient().getServerURI();
    }

    @Override
    public IMqttDeliveryToken[] getPendingDeliveryTokens() {
        return this.getMqttClient().getPendingDeliveryTokens();
    }

    @Override
    public void setManualAcks(boolean manualAcks) {
        this.getMqttClient().setManualAcks(manualAcks);
    }

    @Override
    public void reconnect() throws MqttException {
        this.getMqttClient().reconnect();
    }

    @Override
    public void messageArrivedComplete(int messageId, int qos) throws MqttException {
        this.getMqttClient().messageArrivedComplete(messageId, qos);
    }

}
